package com.iflytek.controller;

import com.alibaba.fastjson.JSON;
import com.iflytek.bean.MonitorInfo;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.Properties;

public class _03TaoPaiCarController {
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Violation {
        private int id;
        private String car;
        private String violation;
        private Long createTime;
    }


    public static void main(String[] args) throws Exception {

        // 1. env-准备环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);

        // 2. source-加载数据
        // 2. source-加载数据  读取kafka中的消息
        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "hadoop102:9092,hadoop103:9092,hadoop104:9092");
        properties.setProperty("group.id", "g1");
        FlinkKafkaConsumer<String> kafkaSource = new FlinkKafkaConsumer<String>("topic-car", new SimpleStringSchema(), properties);
        DataStreamSource<String> dataStreamSource = env.addSource(kafkaSource);

        // 3. transformation-数据处理转换
        // 进行转换  将数据变为一个个的对象
        DataStream<MonitorInfo> mapStream = dataStreamSource.map(new MapFunction<String, MonitorInfo>() {
            @Override
            public MonitorInfo map(String jsonStr) throws Exception {

                // 通过代码可以观察到，json中即使字段不太一致，只要 action_time  类中的字段 actionTime
                MonitorInfo monitorInfo = JSON.parseObject(jsonStr, MonitorInfo.class);

                return monitorInfo;
            }
        });
        // 3. transformation-数据处理转换
        // 按照 MonitorInfo 对象的车牌号进行按键分组，相同车牌号的数据会被分到同一组。
        SingleOutputStreamOperator<Violation> resultDs = mapStream.keyBy(monitorInfo -> monitorInfo.getCar()).flatMap(new RichFlatMapFunction<MonitorInfo, Violation>() {

            // 定义一个 ValueState 类型的状态变量 valueState，用于存储每个车牌号对应的上一条 MonitorInfo 数据。
            // ValueState 是 Flink 提供的一种状态管理机制，能在任务失败重启后恢复状态。
            ValueState<MonitorInfo> valueState;
            // open用于初始化，只执行一次
            @Override
            public void open(Configuration parameters) throws Exception {
                // ValueStateDescriptor：用于描述状态，指定状态名称为 vs1，状态类型为 MonitorInfo。
                ValueStateDescriptor<MonitorInfo> stateDescriptor = new ValueStateDescriptor<MonitorInfo>("vs1", MonitorInfo.class);
                // getRuntimeContext().getState(stateDescriptor)：从运行时上下文中获取 ValueState 实例。
                valueState = getRuntimeContext().getState(stateDescriptor);
            }

            @Override
            public void flatMap(MonitorInfo monitorInfo, Collector<Violation> collector) throws Exception {

                // 根据车牌号，获取上一个汽车的 时间，还有 摄像头的id
                MonitorInfo _monitorInfo = valueState.value();
                // 将当前 MonitorInfo 数据更新到状态中，以便后续处理使用。
                valueState.update(monitorInfo);
                if (_monitorInfo != null && monitorInfo.getActionTime() - _monitorInfo.getActionTime() < 10 && monitorInfo.getMonitorId() != _monitorInfo.getMonitorId()) {
                    Violation violation = new Violation(0, monitorInfo.getCar(), "涉嫌套牌", System.currentTimeMillis());
                    collector.collect(violation);
                }

            }
        });
        resultDs.print();
        // 4. sink-数据输出
        resultDs.addSink(JdbcSink.sink("INSERT INTO `t_violation_list` VALUES (?,?,?)", new JdbcStatementBuilder<Violation>() {
            @Override
            public void accept(PreparedStatement ps, Violation value) throws SQLException {
                ps.setString(1, value.getCar());
                ps.setString(2, value.getViolation());
                ps.setLong(3, value.getCreateTime());
            }
        }, JdbcExecutionOptions.builder().withBatchSize(1).withBatchIntervalMs(5000).build(), new JdbcConnectionOptions.JdbcConnectionOptionsBuilder().withDriverName("com.mysql.cj.jdbc.Driver").withPassword("12345678").withUrl("jdbc:mysql://hadoop102:3306/flink_project").withUsername("root").build()));

        // 5. execute-执行
        env.execute();
    }
}
